package cn.com.ytst.data.integration.components.reader.mq;

import cn.com.ytst.data.integration.core.channel.PipleChannel;
import cn.com.ytst.data.integration.core.plugin.Reader;
import cn.com.ytst.data.integration.util.Constants;
import com.alibaba.datax.common.util.Configuration;
import com.rabbitmq.client.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.util.HashMap;
import java.util.Map;


/**
 * @ClassName RabbitmqReader
 * @Deacription
 * @Author Administrator
 * @Date 2020/1/4 0004 20:10
 * @Version 1.0
 **/
public class RabbitmqReader extends Reader {

    private Logger logger = LoggerFactory.getLogger(this.getClass());

    private ConnectionFactory factory = null;

    private Consumer consumer;

    private Channel channel;

    private String queue;


    private String consumerTag;

    private PipleChannel pipleChannel;

    public RabbitmqReader(){
    }


    @Override
    public void init() {
       // this.setAbstractEvent(abstractEvent);
        pipleChannel = this.getPipleChannel();
        Configuration configuration = this.getPluginJobConf();
        this.factory = new ConnectionFactory();
        String host = configuration.getString(Constants.MQ_HOST_TAG,Constants.DEFAULT_VALUE);
        String username = configuration.getString(Constants.MQ_USERNAME_TAG,Constants.DEFAULT_VALUE);
        String password = configuration.getString(Constants.MQ_PWD_TAG,Constants.DEFAULT_VALUE);
        String port = configuration.getString(Constants.MQ_PORT_TAG,"5672");
        queue = configuration.getString(Constants.MQ_QUEUE_TAG,Constants.DEFAULT_VALUE);
        factory.setVirtualHost("/");
        factory.setHost(host);
        factory.setPassword(password);
        factory.setUsername(username);
        factory.setPort(Integer.parseInt(port));
        factory.setRequestedChannelMax(10);
        Connection connection = null;
        Map<String, Object> args = new HashMap<>();
        args.put("x-max-length", 10000);
        try {
            connection = factory.newConnection();
            channel = connection.createChannel();
            channel.queueDeclare(queue, false, false, false, args);


        } catch (Exception e) {
            logger.error("发生错误,错误原因:{}",e);
        }

    }



    @Override
    public void startReader() {
         consumer = new DefaultConsumer(channel) {

            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) {
                String message = new String(body);
                pipleChannel.doPush(message);
            }
        };
        try {
            consumerTag =  channel.basicConsume(this.queue, false, consumer);
        } catch (IOException e) {
            logger.error("发生错误,错误原因:{}",e);
        }
    }

    @Override
    public void destroy() {
        try {
            channel.basicCancel(consumerTag);
            channel.close();
            channel.abort();

        } catch (IOException e) {
            e.printStackTrace();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }


}
